package com.schedule.ticker;

import java.util.List;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.LockSupport;

public class TimeWheel {

    private final Ticker ticker;

    private final MpscTaskQueue[] wheel;

    private volatile Long startTime;

    private Long TICK_TIME = 100L;

    private final CountDownLatch startTimeLatch;

    private final ExecutorService executor;

    // 用标记管理Ticker线程的启动关闭
    private final AtomicBoolean started;

    public TimeWheel(){
        wheel = new MpscTaskQueue[10];
        ticker = new Ticker();
        started = new AtomicBoolean(false);
        startTimeLatch = new CountDownLatch(1);
        executor = Executors.newFixedThreadPool(6);

        for(int i=0;i<10;i++){
//            wheel[i] = new Slot();
            // 用并发无锁队列 代替 我们简单的链表结构
            wheel[i] = new MpscTaskQueue();
        }
    }

    public void addDelayTask(Runnable runnable,long delayMs){
        start();
        DelayTask delayTask = new DelayTask(runnable,delayMs);
        int index = Math.toIntExact(((delayTask.deadline - startTime) / 100 ) % wheel.length);
        MpscTaskQueue slot = wheel[index];
        slot.pushTask(delayTask);
    }

    private void start(){
        if(!started.compareAndSet(false,true)){
            ticker.start();
            // 保证在主线程使用startTime之前，startTime被Ticker线程进行赋值
            while (startTime == 0L){
                try {
                    startTimeLatch.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    public void stop(){
        if (started.compareAndSet(true,false)) {
            LockSupport.unpark(ticker);
        }
    }

    public static class DelayTask{
         final Runnable runnable;
         long deadline;
         DelayTask next;
         DelayTask pre;

        public DelayTask(Runnable runnable, long deadline) {
            this.runnable = runnable;
            this.deadline = deadline;
        }
    }

    public class Slot{

        DelayTask head;
        DelayTask tail;

        public void runWithDeadline(long tickTime){
            DelayTask current = head;
            while(current != null){
                DelayTask next = current.next;
                if(current.deadline <= tickTime){
                    removeTask(current);
                    // 这里ticker线程既充当了触发器，又充当了执行器
//                    current.runnable.run();
                    // 交给线程池执行
                    executor.execute(current.runnable);
                }
                current = next;
            }
        }

        private void removeTask(DelayTask task){
            if (task.pre != null) {
                task.pre.next =task.next;
            }
            if(task.next != null){
                task.next.pre = task.pre;
            }
            if(task == head){
                head = task.next;
            }
            if(task == tail){
                tail = task.pre;
            }
            task.next = null;
            task.pre = null;
        }

        public void pushDelayTask(DelayTask delayTask){
            if(head == null){
                head = tail = delayTask;
            }else{
                tail.next = delayTask;
                delayTask.pre = tail;
                tail = delayTask;
            }
        }
    }

    public class Ticker extends Thread{
        int tickCount = 0;

        @Override
        public void run() {
            startTime = System.currentTimeMillis();
            startTimeLatch.countDown();
            while (started.get()){
                long tickTime = startTime + (tickCount + 1) * TICK_TIME;
                while(System.currentTimeMillis() <= tickTime){
                    LockSupport.parkUntil(tickTime);
                    if(!started.get()){
                        return;
                    }
                }
                int index = tickCount % wheel.length;
//                Slot slot = wheel[index];
//                slot.runWithDeadline(tickTime);
                MpscTaskQueue queue = wheel[index];
                List<Runnable> runnables = queue.removeAndReturnShouldRun(tickTime);
                runnables.forEach(executor::execute);
                tickCount++;
            }
        }
    }
}
